package org.atlantis.datasink.server.buildin;

import java.net.SocketAddress;
import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.atlantis.datasink.codec.CodecOption;
import org.jboss.netty.bootstrap.ClientBootstrap;
import org.jboss.netty.channel.ChannelFuture;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public final class ActiveConnectManager {

	private static final Logger log = LoggerFactory.getLogger(ActiveConnectManager.class);

	private static class Holder {
		static ActiveConnectManager instance = new ActiveConnectManager();
	}

	public static ActiveConnectManager getInstance() {
		return Holder.instance;
	}

	private CommandSender sender = new CommandSender();
	private ConcurrentMap<SocketAddress, Connector> clientMap = new ConcurrentHashMap<SocketAddress, Connector>();
	private Queue<SocketAddress> failedSet = new LinkedList<SocketAddress>();
	private Thread reconnectWorker;
	private volatile boolean isAlive = true;

	private Lock lock = new ReentrantLock();
	private Condition empty = lock.newCondition();

	private ActiveConnectManager() {
		reconnectWorker = new Thread(new Runnable() {
			@Override
			public void run() {
				reconnect();
			}
		}, "Reconnector-Worker");
		reconnectWorker.setDaemon(true);
		reconnectWorker.start();
	}

	public void connect(ClientBootstrap bootstrap) {
		checkAlive();
		Connector conn = new Connector(bootstrap);
		clientMap.put(conn.addr, conn);
		connect(conn, false);
	}

	public void waitToReconnect(SocketAddress addr) {
		checkAlive();
		lock.lock();
		try {
			failedSet.offer(addr);
			empty.signalAll();
		} finally {
			lock.unlock();
		}
		sender.unregisterCommand(addr);
	}

	private void connect(Connector conn, boolean isRedo) {
		ChannelFuture future = conn.bootstrap.connect();
		SocketAddress addr = conn.addr;
		future.awaitUninterruptibly(isRedo ? 5000L : 15000L);
		if (!future.isSuccess()) {
			if (isRedo) log.debug("Can not connect to server: " + addr);
			else log.error("Can not connect to server: " + addr);
			lock.lock();
			try {
				failedSet.offer(addr);
				empty.signalAll();
			} finally {
				lock.unlock();
			}
			return;
		}
		sender.registerCommand(conn.codecOpt, future.getChannel());
	}

	private void reconnect() {
		while (!Thread.currentThread().isInterrupted()) {
			SocketAddress addr = null;
			lock.lock();
			try {
				addr = failedSet.poll();
				if (addr == null) empty.await();
			} catch (InterruptedException e) {
				Thread.currentThread().interrupt();
			} finally {
				lock.unlock();
			}
			Connector conn = clientMap.get(addr);
			if (conn == null) continue;
			connect(conn, true);
		}
	}

	public void destory() {
		isAlive = false;
		sender.destory();
		clientMap.clear();
		failedSet.clear();
		empty.signalAll();
	}

	private void checkAlive() {
		if (!isAlive) throw new IllegalStateException("ActiveConnectManager is closed.");
	}

	private static class Connector {
		CodecOption codecOpt;
		ClientBootstrap bootstrap;
		SocketAddress addr;

		Connector(ClientBootstrap bootstrap) {
			this.codecOpt = (CodecOption) bootstrap.getOption("codecOption");
			this.bootstrap = bootstrap;
			this.addr = (SocketAddress) bootstrap.getOption("remoteAddress");
		}
	}
}
